package test2.hash;

import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;

import org.nlpcn.commons.lang.util.MD5;

/**
 * 一致性Hash算法
 *
 * @param <T>
 *            节点类型
 */
public class ConsistentHash<T> {
	/** Hash计算对象，用于自定义hash算法 */
	HashFunc hashFunc;
	/** 复制的节点个数 */
	private final int numberOfReplicas;
	/** 一致性Hash环 */
	private final SortedMap<Integer, Node> circle = new TreeMap<Integer, Node>();
	
	private final Map<String, Map<String, Object>> totalMap = new HashMap<>();

	public int getNumberOfReplicas() {
		return numberOfReplicas;
	}

	public SortedMap<Integer, Node> getCircle() {
		return circle;
	}

	/**
	 * 构造，使用Java默认的Hash算法
	 * 
	 * @param numberOfReplicas
	 *            复制的节点个数，增加每个节点的复制节点有利于负载均衡
	 * @param nodes
	 *            节点对象
	 */
	public ConsistentHash(int numberOfReplicas, Collection<T> nodes) {
		this.numberOfReplicas = numberOfReplicas;
		this.hashFunc = new HashFunc() {
			@Override
			public Integer hash(Object key) {
//				return key.hashCode();
				return MD5.code(key.toString()).hashCode();
			}
		};

		// 初始化节点
		for (T node : nodes) {
			add(node);
		}
	}

	/**
	 * 构造
	 * 
	 * @param hashFunc
	 *            hash算法对象
	 * @param numberOfReplicas
	 *            复制的节点个数，增加每个节点的复制节点有利于负载均衡
	 * @param nodes
	 *            节点对象
	 */
	public ConsistentHash(HashFunc hashFunc, int numberOfReplicas, Collection<T> nodes) {
		this.numberOfReplicas = numberOfReplicas;
		this.hashFunc = hashFunc;
		// 初始化节点
		for (T node : nodes) {
			add(node);
		}
	}

	/**
	 * 增加节点<br>
	 * 每增加一个节点，就会在闭环上增加给定复制节点数<br>
	 * 例如复制节点数是2，则每调用此方法一次，增加两个虚拟节点，这两个节点指向同一Node
	 * 由于hash算法会调用node的toString方法，故按照toString去重
	 * 
	 * @param node
	 *            节点对象
	 */
	public void add(T node, boolean flag) {
		if(flag) {
			System.out.println("----迁移开始----");
			move(node);
			System.out.println("----迁移结束----");
		}
		for (int i = 0; i < numberOfReplicas; i++) {
			circle.put(hashFunc.hash(node.toString() + i), new Node(node, i));
			totalMap.put(node.toString() + i, new HashMap<String, Object>());
		}
	}
	
	public void add(T node) {
		add(node, false);
	}

	/**
	 * 移除节点的同时移除相应的虚拟节点
	 * 
	 * @param node
	 *            节点对象
	 */
	public void remove(T node) {
		System.out.println("----迁移开始----");
		move(node);
		System.out.println("----迁移结束----");
		for (int i = 0; i < numberOfReplicas; i++) {
			circle.remove(hashFunc.hash(node.toString() + i));
		}
		totalMap.remove(node);
	}
	
	/**
	 * 数据迁移
	 * @param node
	 */
	public void move(T node) {
		Map<Node, Node> tempMap = new HashMap<>();
		//确立初步对应关系
		for (int i = 0; i < numberOfReplicas; i++) {
			
			int hash = hashFunc.hash(node.toString() + i);
			SortedMap<Integer, Node> tailMap = circle.tailMap(hash);
			int size = tailMap.size();
			Set<Integer> keys = tailMap.keySet();
			Iterator<Integer> iter = keys.iterator();
			int key1 = iter.next();
			Node node1 = circle.get(key1);
//			System.out.println("firstValue = " + node1);
			
			if(size==1) {
				SortedMap<Integer, Node> headMap = circle.headMap(hash);
//				System.out.println(headMap);
				Node node2 = circle.get(headMap.firstKey());
//				System.out.println("uniqueValue = " + node1);
				
				tempMap.put(node1, node2);
				
			} else {
				
				
				int key2 = iter.next();
				Node node2 = circle.get(key2);
//				System.out.println("secondValue = " + node2);
				
				tempMap.put(node1, node2);
			}
		}
		
		//确立真正对应关系
		for(Map.Entry<Node, Node> entry : tempMap.entrySet()) {
			Node key = entry.getKey();
			Node value = entry.getValue();
			if(key.getNode().equals(value.getNode())) {
				recur(key, tempMap.get(value), tempMap);
			} else {
				tempMap.put(key, value);
			}
		}
		
		//迁移
		for(Map.Entry<Node, Node> entry : tempMap.entrySet()) {
			Node key = entry.getKey();
			Node value = entry.getValue();
			
			Map<String, Object> keyMap = totalMap.get(key.getKey());
			Map<String, Object> valueMap = totalMap.get(value.getKey());
			
			valueMap.putAll(keyMap);
		}
		
		//删除
		for(Map.Entry<Node, Node> entry : tempMap.entrySet()) {
			Node key = entry.getKey();
			totalMap.remove(key.getKey());
		}
		
	}
	
	private void recur(Node node1, Node node2, Map<Node, Node> map) {
		if(node1.getNode().equals(node2.getNode())) {
			recur(node1, map.get(node2), map);
		} else {
			map.put(node1, node2);
		}
	}
	
	/**
	 * 获得一个最近的顺时针节点
	 * 
	 * @param key
	 *            为给定键取Hash，取得顺时针方向上最近的一个虚拟节点对应的实际节点
	 * @return 节点对象
	 */
	public Node get(Object key) {
		if (circle.isEmpty()) {
			return null;
		}
		int hash = hashFunc.hash(key);
		if (!circle.containsKey(hash)) {
			SortedMap<Integer, Node> tailMap = circle.tailMap(hash);

			// 返回此映射的部分视图，其键大于等于 hash
			hash = tailMap.isEmpty() ? circle.firstKey() : tailMap.firstKey();
		}
		// 正好命中
		return circle.get(hash);
	}
	
	/**
	 * 获取
	 * @param key
	 * @return
	 */
	public Object getObj(String key) {
		Node node = get(key);
		Map<String, Object> valueMap = totalMap.get(node.getKey());
		if(valueMap==null) {
			System.out.println(node + "节点被删除");
			return null;
		}
		return node + "--->" + valueMap.get(key);
	}
	
	/**
	 * 存放
	 * @param key
	 * @param val
	 * @return
	 */
	public Node putObj(String key, Object val) {
		Node node = get(key);
		totalMap.get(node.getKey()).put(key, val);
		return node;
	}
	
	public static void main(String [] args) { 
        HashSet< String> set = new HashSet< String>();  
        set.add( "A" );  
        set.add( "B" );  
//        set.add( "C" );  
//        set.add( "D" );  

        ConsistentHash<String> consistentHash = new ConsistentHash<String>(3, set);  

        System.out.println(consistentHash.getCircle());
        
        System.out.println(consistentHash.putObj("newsa", "辽宁胜浙江"));
        System.out.println(consistentHash.putObj("newsb", "新疆胜北京"));
        System.out.println(consistentHash.putObj("newsc", "广东胜山东"));
        System.out.println(consistentHash.putObj("newsd", "四川胜广厦"));
        
        consistentHash.add("E", true);
        
        System.out.println(consistentHash.getCircle());
        
        System.out.println(consistentHash.getObj("newsa"));
        System.out.println(consistentHash.getObj("newsb"));
        System.out.println(consistentHash.getObj("newsc"));
        System.out.println(consistentHash.getObj("newsd"));
        
//        news1 = consistentHash.get("news1");
//        map.put("news1", news1);
//        news2 = consistentHash.get("news2");
//        map.put("news2", news2);
//        news3 = consistentHash.get("news3");
//        map.put("news3", news3);
//        news4 = consistentHash.get("news4");
//        map.put("news4", news4);
//        System.out.println(map);
        
        
	}  

	public static void showServer(Map<String , Integer> map) {  
		for (Entry<String, Integer> m : map.entrySet()) {  
              System. out .println("服务器 " + m.getKey() + "----" + m.getValue() + "个" );  
        }  
	}  
}